package com.kmxd.ams.app.es;

import static cn.easyes.common.constants.BaseEsConstants.DEFAULT_ES_ID_NAME;

import cn.easyes.annotation.rely.Analyzer;
import cn.easyes.annotation.rely.FieldType;
import cn.easyes.common.constants.BaseEsConstants;
import cn.easyes.common.utils.ArrayUtils;
import cn.easyes.common.utils.ExceptionUtils;
import cn.easyes.core.conditions.index.LambdaEsIndexWrapper;
import cn.easyes.core.conditions.select.LambdaEsQueryWrapper;
import cn.easyes.core.core.BaseEsMapper;
import cn.easyes.core.toolkit.EntityInfoHelper;
import cn.easyes.core.toolkit.FieldUtils;
import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.BooleanUtil;
import cn.hutool.core.util.NumberUtil;
import cn.hutool.core.util.ObjectUtil;
import com.baomidou.mybatisplus.core.toolkit.ReflectionKit;
import com.baomidou.mybatisplus.core.toolkit.StringUtils;
import com.baomidou.mybatisplus.extension.toolkit.SqlHelper;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.Maps;
import com.kmxd.ams.client.archive.IArchiveFieldSvc;
import com.kmxd.ams.client.archive.dto.ArchiveFieldQueryDTO;
import com.kmxd.ams.client.archive.vo.ArchiveFieldListVO;
import com.kmxd.ams.client.es.IServiceEs;
import com.kmxd.ams.core.common.constant.ArchConst;
import com.kmxd.ams.core.common.constant.ErrorCode;
import com.kmxd.ams.core.common.constant.EsConst;
import com.kmxd.ams.core.common.constant.RedissonConst;
import com.kmxd.ams.core.exception.BizTipException;
import com.kmxd.ams.infra.arrange.base.BaseBO;
import com.kmxd.ams.infra.search.entity.DocumentEs;
import com.kmxd.ams.infra.search.entity.EFileRecordEs;
import com.kmxd.ams.infra.tmpl.dto.ArchTmplBO;
import com.kmxd.ams.infra.utils.EsFieldUtil;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;
import org.apache.commons.compress.utils.Lists;
import org.apache.ibatis.logging.Log;
import org.apache.ibatis.logging.LogFactory;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.client.indices.GetIndexResponse;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.core.TimeValue;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptType;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;

/**
 * IService 实现类（ 泛型：M 是 mapper 对象，T 是实体 ）
 *
 * @param <M>
 * @param <T>
 */
@SuppressWarnings("unchecked")
public class ServiceEsImpl<M extends BaseEsMapper<T>, T> implements IServiceEs<T> {

  protected Log log = LogFactory.getLog(getClass());

  @Autowired protected M baseMapper;
  @Autowired protected RestHighLevelClient restHighLevelClient;
  protected Class<T> entityClass = currentModelClass();
  protected Class<M> mapperClass = currentMapperClass();
  @Autowired private IArchiveFieldSvc archiveFieldSvc;

  @Value("${easy-es.global-config.replicas:2}")
  private Integer defaultReplicas;

  @Value("${easy-es.global-config.shards:3}")
  private Integer defaultShards;

  @Value("${easy-es.global-config.maxWordsSize:3000}")
  private Integer maxWordsSize;

  @Autowired private RedissonClient redissonClient;

  @Override
  public boolean saveBatch(Collection<T> entityList, int batchSize, String indexName) {
    if (ObjectUtil.isEmpty(entityList)) {
      return true;
    }
    Integer integer = baseMapper.insertBatch(entityList, indexName);
    return ObjectUtil.equals(integer, entityList.size());
  }

  @Override
  public boolean updateByQuery(
      T entity, LambdaEsQueryWrapper<T> queryWrapper, String... indexName) {
    try {
      UpdateByQueryRequest request = new UpdateByQueryRequest(indexName);
      Map<String, Object> map = BeanUtil.beanToMap(entity, false, true);
      StringBuilder scriptBuilder = new StringBuilder();
      map.forEach(
          (key, value) -> {
            scriptBuilder.append("ctx._source." + key + "='" + value + "';");
          });
      Script script =
          new Script(ScriptType.INLINE, "painless", scriptBuilder.toString(), ImmutableMap.of());
      request.setScript(script);
      request.setTimeout(TimeValue.timeValueMinutes(3L));
      request.setRefresh(true);
      String source = baseMapper.getSource(queryWrapper);
      ObjectMapper mapper = new ObjectMapper();
      JsonNode tree = mapper.readTree(source);
      String queryStr = tree.get("query").toString();
      request.setQuery(QueryBuilders.wrapperQuery(queryStr));
      BulkByScrollResponse response =
          restHighLevelClient.updateByQuery(request, RequestOptions.DEFAULT);
      long count = baseMapper.selectCount(queryWrapper);
      long updated = response.getStatus().getUpdated();
      return ObjectUtil.equals(updated, count);
    } catch (Exception e) {
      log.error(e.getMessage(), e);
    }
    return false;
  }

  @Override
  public boolean updateBatchById(Collection<T> entityList, String indexName) {
    if (ObjectUtil.isEmpty(entityList)) {
      return true;
    }
    Integer integer = baseMapper.updateBatchByIds(entityList, indexName);
    return ObjectUtil.equals(integer, entityList.size());
  }

  @Override
  public boolean updateBatchById(Collection<T> entityList, int batchSize, String indexName) {
    if (ObjectUtil.isEmpty(entityList)) {
      return true;
    }
    AtomicLong count = new AtomicLong(0L);
    Iterables.partition(entityList, batchSize)
        .forEach(
            batch -> {
              Integer integer = baseMapper.updateBatchByIds(batch, indexName);
              count.getAndAdd(integer);
            });
    return ObjectUtil.equals(count.get(), entityList.size());
  }

  @Override
  public boolean existsIndex(BaseBO req) {
    return baseMapper.existsIndex(req.getTableId());
  }

  @Override
  public boolean createIndex(BaseBO req, Boolean cleanOldIndex) {
    return createIndex(req, cleanOldIndex, defaultShards, defaultReplicas);
  }

  @Override
  public boolean createIndex(BaseBO req, Boolean cleanOldIndex, Integer shards, Integer replicas) {
    RLock lock = redissonClient.getLock(RedissonConst.KEY_PREFIX + req.getTableId());
    try {
      boolean isLocked =
          lock.tryLock(RedissonConst.WAIT_TIME, RedissonConst.LEASE_TIME, TimeUnit.SECONDS);
      // 未拿到锁,直接返回
      if (!isLocked) {
        return false;
      }
      LambdaEsIndexWrapper<T> wrapper = new LambdaEsIndexWrapper<>();
      wrapper.indexName(req.getTableId());
      // 检测index是否已经存在,已存在先删除之前的index
      boolean existsIndex = existsIndex(req);
      if (existsIndex) {
        if (BooleanUtil.isTrue(cleanOldIndex)) {
          removeIndex(req);
        } else {
          return true;
        }
      }
      ArchiveFieldQueryDTO qry = new ArchiveFieldQueryDTO();
      qry.setTableId(req.getTableId());
      qry.setArchTypeId(req.getArchTypeId());
      List<ArchiveFieldListVO> list = archiveFieldSvc.selectList(qry);
      if (ObjectUtil.isEmpty(list)) {
        throw BizTipException.instance(ErrorCode.VALIDATE_ERROR, "未找到元数据信息");
      }
      Map<String, Object> fieldMapping = Maps.newHashMap();
      list.forEach(
          field -> {
            FieldType fieldType = EsFieldUtil.convertType(field.getDbType());
            // 特殊类型处理
            if (field.getDbFieldKey().equals(ArchConst.DEF_STATUS)) {
              fieldType = FieldType.INTEGER;
            }
            if (field.getDbFieldKey().equals(ArchConst.DEF_ID)) {
              fieldMapping.put(
                  StringUtils.underlineToCamel(field.getDbFieldKey()),
                  ImmutableMap.of(BaseEsConstants.TYPE, FieldType.LONG.getType()));
            } else if (ImmutableSet.of(ArchConst.ARCH_CODE_DB, ArchConst.FIELD_TITLE)
                .contains(field.getDbFieldKey())) {
              // 分词类型
              fieldMapping.put(
                  StringUtils.underlineToCamel(field.getDbFieldKey()),
                  ImmutableMap.of(
                      BaseEsConstants.TYPE,
                      FieldType.TEXT.getType(),
                      BaseEsConstants.ANALYZER,
                      Analyzer.IK_MAX_WORD,
                      EsConst.MP_FIELDS,
                      ImmutableMap.of(
                          FieldType.KEYWORD.getType(),
                          ImmutableMap.of(BaseEsConstants.TYPE, FieldType.KEYWORD.getType()))));
            } else if (ImmutableSet.of(ArchConst.FIELD_YEAR).contains(field.getDbFieldKey())) {
              // 分词类型
              fieldMapping.put(
                  StringUtils.underlineToCamel(field.getDbFieldKey()),
                  ImmutableMap.of(BaseEsConstants.TYPE, FieldType.KEYWORD.getType()));
            } else {
              // 时间，设置格式化
              if (fieldType.equals(FieldType.DATE)) {
                String format = "yyyy-MM-dd HH:mm:ss||yyyy-MM-dd||epoch_millis";
                fieldMapping.put(
                    StringUtils.underlineToCamel(field.getDbFieldKey()),
                    ImmutableMap.of(
                        BaseEsConstants.TYPE,
                        fieldType.getType(),
                        BaseEsConstants.FORMAT,
                        format,
                        EsConst.MP_FIELDS,
                        ImmutableMap.of(
                            FieldType.KEYWORD.getType(),
                            ImmutableMap.of(BaseEsConstants.TYPE, FieldType.KEYWORD.getType()))));
              } else {
                fieldMapping.put(
                    StringUtils.underlineToCamel(field.getDbFieldKey()),
                    ImmutableMap.of(BaseEsConstants.TYPE, fieldType.getType()));
              }
            }
          });

      // 添加单位信息
      fieldMapping.put(
          ArchConst.COM_ID, ImmutableMap.of(BaseEsConstants.TYPE, FieldType.KEYWORD.getType()));
      fieldMapping.put(
          ArchConst.COM_NAME, ImmutableMap.of(BaseEsConstants.TYPE, FieldType.KEYWORD.getType()));
      fieldMapping.put(
          ArchConst.TABLE_ID, ImmutableMap.of(BaseEsConstants.TYPE, FieldType.KEYWORD.getType()));
      fieldMapping.put(
          ArchConst.ARCH_TYPE_ID,
          ImmutableMap.of(BaseEsConstants.TYPE, FieldType.KEYWORD.getType()));
      fieldMapping.put(
          ArchConst.ARCH_TYPE_NAME,
          ImmutableMap.of(BaseEsConstants.TYPE, FieldType.KEYWORD.getType()));
      fieldMapping.put(
          ArchConst.ARCH_MODEL, ImmutableMap.of(BaseEsConstants.TYPE, FieldType.KEYWORD.getType()));
      fieldMapping.put(
          ArchConst.ARCH_MODEL_NAME,
          ImmutableMap.of(BaseEsConstants.TYPE, FieldType.KEYWORD.getType()));
      //    wrapper.mapping(
      //        ArchConst.ES_AMS_FILE_CONTENT,
      //        FieldType.KEYWORD_TEXT,
      //        Analyzer.IK_MAX_WORD,
      //        Analyzer.IK_MAX_WORD);
      //    wrapper.mapping(
      //        ArchConst.ES_AMS_FILE_TITLE,
      //        FieldType.KEYWORD_TEXT,
      //        Analyzer.IK_MAX_WORD,
      //        Analyzer.IK_MAX_WORD);

      // 添加权限信息

      //     文件门类
      if (ObjectUtil.contains(req.getTableId(), ArchTmplBO.DOCUMENT)) {
        fieldMapping.putAll(buildOcrMapping());
      }
      wrapper.mapping(ImmutableMap.of(BaseEsConstants.PROPERTIES, fieldMapping));
      // 最大返回数据
      wrapper.settings(
          Settings.builder().put("index.max_result_window", Integer.MAX_VALUE).build());
      // 设置分片及副本信息,3个shards,2个replicas,可缺省
      wrapper.settings(
          Optional.ofNullable(shards).orElse(defaultShards),
          Optional.ofNullable(replicas).orElse(defaultReplicas));
      return baseMapper.createIndex(wrapper);
    } catch (Exception e) {
      log.error(e.getMessage(), e);
      return false;
    } finally {
      lock.unlock();
    }
  }

  private Map<String, Object> buildOcrMapping() {
    return ImmutableMap.of(
        FieldUtils.val(DocumentEs::getRecords),
        ImmutableMap.of(
            BaseEsConstants.TYPE,
            FieldType.NESTED.getType(),
            BaseEsConstants.PROPERTIES,
            // 字段信息
            ImmutableMap.of(
                // 电子文档id
                FieldUtils.val(EFileRecordEs::getId),
                ImmutableMap.of(BaseEsConstants.TYPE, FieldType.KEYWORD.getType()),
                // 文件大小
                FieldUtils.val(EFileRecordEs::getSize),
                ImmutableMap.of(BaseEsConstants.TYPE, FieldType.KEYWORD.getType()),
                // 文件扩展名
                FieldUtils.val(EFileRecordEs::getExt),
                ImmutableMap.of(BaseEsConstants.TYPE, FieldType.KEYWORD.getType()),
                // 标题
                FieldUtils.val(EFileRecordEs::getTitle),
                ImmutableMap.of(
                    BaseEsConstants.TYPE,
                    FieldType.TEXT.getType(),
                    BaseEsConstants.ANALYZER,
                    Analyzer.IK_MAX_WORD,
                    EsConst.MP_FIELDS,
                    ImmutableMap.of(
                        FieldType.KEYWORD.getType(),
                        ImmutableMap.of(BaseEsConstants.TYPE, FieldType.KEYWORD.getType()))),
                // ocr内容字段
                FieldUtils.val(EFileRecordEs::getOcrContent),
                ImmutableMap.of(
                    BaseEsConstants.TYPE,
                    FieldType.TEXT.getType(),
                    BaseEsConstants.ANALYZER,
                    Analyzer.IK_MAX_WORD,
                    EsConst.MP_FIELDS,
                    ImmutableMap.of(
                        FieldType.KEYWORD.getType(),
                        ImmutableMap.of(
                            BaseEsConstants.TYPE,
                            FieldType.KEYWORD.getType(),
                            EsConst.MP_IGNORE_ABOVE,
                            maxWordsSize))))));
  }

  @Override
  public boolean removeIndex(BaseBO req) {
    return baseMapper.deleteIndex(req.getTableId());
  }

  @Override
  public M getBaseMapper() {
    return baseMapper;
  }

  @Override
  public Class<T> getEntityClass() {
    return entityClass;
  }

  /**
   * 判断数据库操作是否成功
   *
   * @param result 数据库操作返回影响条数
   * @return boolean
   * @deprecated 3.3.1
   */
  @Deprecated
  protected boolean retBool(Integer result) {
    return SqlHelper.retBool(result);
  }

  protected Class<M> currentMapperClass() {
    return (Class<M>)
        ReflectionKit.getSuperClassGenericType(this.getClass(), ServiceEsImpl.class, 0);
  }

  protected Class<T> currentModelClass() {
    return (Class<T>)
        ReflectionKit.getSuperClassGenericType(this.getClass(), ServiceEsImpl.class, 1);
  }

  @Override
  public List<Long> checkDataExist(Collection<Long> ids, String tableId) {
    if (ObjectUtil.isEmpty(ids)) {
      return Lists.newArrayList();
    }
    // 构造查询参数
    List<String> stringIdList = ids.stream().map(Object::toString).collect(Collectors.toList());
    SearchRequest searchRequest = new SearchRequest(tableId);
    Optional.ofNullable(EntityInfoHelper.getEntityInfo(entityClass).getRouting())
        .ifPresent(searchRequest::routing);
    SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
    sourceBuilder.query(QueryBuilders.termsQuery(DEFAULT_ES_ID_NAME, stringIdList));
    sourceBuilder.size(ids.size());
    sourceBuilder.fetchField(DEFAULT_ES_ID_NAME);
    searchRequest.source(sourceBuilder);

    // 请求es获取数据
    SearchHit[] searchHitArray = getSearchHitArray(searchRequest);
    if (ArrayUtils.isEmpty(searchHitArray)) {
      return new ArrayList<>(0);
    }

    // 批量解析数据
    return Arrays.stream(searchHitArray)
        .map(searchHit -> NumberUtil.parseLong(searchHit.getId()))
        .collect(Collectors.toList());
  }

  private SearchHit[] getSearchHitArray(SearchRequest searchRequest) {
    SearchResponse response;
    try {
      response = baseMapper.search(searchRequest, RequestOptions.DEFAULT);
    } catch (IOException e) {
      throw ExceptionUtils.eee(
          "getSearchHitArray exception,searchRequest:%s", e, searchRequest.toString());
    }
    printResponseErrors(response);
    return parseSearchHitArray(response);
  }

  private void printResponseErrors(SearchResponse searchResponse) {
    if (Objects.nonNull(searchResponse)
        && searchResponse.getShardFailures() != null
        && searchResponse.getShardFailures().length > 0) {
      String errorMsg = searchResponse.getShardFailures()[0].toString();
      throw ExceptionUtils.eee("es响应出错，search response failed ,failedShards: " + errorMsg);
    }
  }

  private SearchHit[] parseSearchHitArray(SearchResponse searchResponse) {
    return Optional.ofNullable(searchResponse)
        .map(SearchResponse::getHits)
        .map(SearchHits::getHits)
        .orElseThrow(() -> ExceptionUtils.eee("parseSearchHitArray exception"));
  }

  @Override
  public String[] checkIndexExist(String[] indexNames) {
    try {
      GetIndexRequest request = new GetIndexRequest("*");
      GetIndexResponse getIndexResponse =
          restHighLevelClient.indices().get(request, RequestOptions.DEFAULT);
      String[] indices = getIndexResponse.getIndices();
      if (ObjectUtil.isEmpty(indices)) {
        return indexNames;
      }
      Set<String> indicesSet = new HashSet<>(Arrays.asList(indices));
      return Arrays.stream(indexNames).filter(indicesSet::contains).toArray(String[]::new);
    } catch (Exception e) {
      log.error(e.getMessage(), e);
      return new String[] {};
    }
  }

  @Override
  public boolean deleteAll() {
    try {
      DeleteIndexRequest request = new DeleteIndexRequest("*");
      AcknowledgedResponse delete =
          restHighLevelClient.indices().delete(request, RequestOptions.DEFAULT);
      return delete.isAcknowledged();
    } catch (Exception e) {
      log.error(e.getMessage(), e);
      return false;
    }
  }
}
